package com.learn.kafka.notify;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Service;

import java.util.Optional;

@Service
public class KafkaConsumerListener implements MessageListener<Integer, String> {

    @KafkaListener(groupId = "lisa", topics = {"seata"})
    void listener(ConsumerRecord<String, String> data) {
        Optional<String> messages = Optional.ofNullable(data.value());
        if (messages.isPresent()) {
            System.out.println("消费者线程：" + Thread.currentThread().getName() + "[ 消息 来自 kafkatopic：" +
                    data.topic() + ",分区：" + data.partition() + " ，委托时间：" + data.timestamp() + "]消息内容如下：");
            System.out.println(data.value());
        }
    }

    @Override
    public void onMessage(ConsumerRecord<Integer, String> integerStringConsumerRecord) {
        System.out.println("收到消息");
        System.out.println(integerStringConsumerRecord.value());
    }
}
